
#include "RtBase.h"
#include "RtThreadManager.h"
#include "RtThreadReactor.h"
#include "RtReactorInterface.h"
/**#include "RtConnectionInterface.h"**/
#include "RtThreadTask.h"
#include "RtAtomicOperationT.h"
#include "RtDnsManager.h"


#if defined (RT_WIN32)
  #include "RtReactorWin32Message.h"
  extern ATOM g_atomRegisterClass;
  extern HINSTANCE g_pReactorWin3Instance;
#elif defined (RT_MACOS)
  #include "RtThreadMacEventPatch.h"
#elif defined (RT_LINUX)
  #include <sys/utsname.h>
  #ifndef RT_PORT_CLIENT
    #include "RtReactorRealTimeSignal.h"
  #endif // RT_PORT_CLIENT
  #ifdef RT_ENABLE_EPOLL
    #include "RtReactorEpoll.h"
  #endif // RT_ENABLE_EPOLL
#endif // RT_WIN32

#ifdef RT_USE_REACTOR_SELECT
  #include "RtReactorSelect.h"
#endif // RT_USE_REACTOR_SELECT

#ifdef RT_WIN32_ENABLE_AsyncGetHostByName
  extern ATOM g_atomDnsRegisterClass;
#endif // RT_WIN32_ENABLE_AsyncGetHostByName
  

static CRtThreadManager *s_pThreadManagerOnlyOne = NULL;
static BOOL s_bThreadManagerOnlyOneByNew = FALSE;

void CRtThreadManager::CleanupOnlyOne()
{
	RT_INFO_TRACE("CRtThreadManager::CleanupOnlyOne");

	if (s_bThreadManagerOnlyOneByNew)
		delete s_pThreadManagerOnlyOne;
	s_pThreadManagerOnlyOne = NULL;
	/**CRtConnectionManager::CleanupInstance();**/
	RT_INFO_TRACE("CRtThreadManager::CleanupOnlyOne, end");
}

CRtThreadManager::CRtThreadManager()
	: m_TTpyeUserThread(TT_USER_DEFINE_BASE)
	, m_pThreadNetwork(NULL)
{
	RT_INFO_TRACE_THIS("CRtThreadManager::CRtThreadManager,");

	RT_ASSERTE(!s_pThreadManagerOnlyOne);
	s_pThreadManagerOnlyOne = this;
}

CRtThreadManager::~CRtThreadManager()
{
	RT_INFO_TRACE_THIS("CRtThreadManager::~CRtThreadManager");
	
	// cleaup instance before delete threads because some instances might use threads.
	CRtCleanUpBase::CleanupAll();

	StopAllThreads(NULL);
	JoinAllThreads();

	ThreadsType tmpThreads;
	{
		CRtMutexGuardT<MutexType> theGuard(m_Mutex);
		m_Threads.swap(tmpThreads);
	}
	ThreadsType::iterator iter = tmpThreads.begin();
	for ( ; iter != tmpThreads.end(); ++iter)
		(*iter)->Destory(RT_OK);

#ifdef RT_WIN32
	if (g_atomRegisterClass) {
		if (!::UnregisterClass(reinterpret_cast<LPCSTR>(g_atomRegisterClass), g_pReactorWin3Instance)) {
			RT_INFO_TRACE("CRtThreadManager::~CRtThreadManager, UnregisterClass() failed!"
				" g_atomRegisterClass=" << g_atomRegisterClass << 
				" g_pReactorWin3Instance=" << g_pReactorWin3Instance << 
				" err=" << ::GetLastError());
		}
		else {
			RT_INFO_TRACE("CRtThreadManager::~CRtThreadManager,"
				" g_atomRegisterClass=" << g_atomRegisterClass << 
				" g_pReactorWin3Instance=" << g_pReactorWin3Instance);
		}
		g_atomRegisterClass = 0;
	}
#endif // RT_WIN32
#ifdef RT_WIN32_ENABLE_AsyncGetHostByName
	if (g_atomDnsRegisterClass) {
		if (!::UnregisterClass(reinterpret_cast<LPCSTR>(g_atomDnsRegisterClass), NULL)) {
			RT_INFO_TRACE("CRtThreadManager::~CRtThreadManager, UnregisterClass() failed!"
				" g_atomDnsRegisterClass=" << g_atomDnsRegisterClass << 
				" err=" << ::GetLastError());
		}
		else {
			RT_INFO_TRACE("CRtThreadManager::~CRtThreadManager,"
				" g_atomDnsRegisterClass=" << g_atomDnsRegisterClass);
		}
		g_atomDnsRegisterClass = 0;
	}
#endif // RT_WIN32_ENABLE_AsyncGetHostByName

	SocketCleanup();
	s_pThreadManagerOnlyOne = NULL;
	/**CRtConnectionManager::CleanupInstance();**/
	m_pThreadNetwork = NULL;
	RT_INFO_TRACE_THIS("CRtThreadManager::~CRtThreadManager end");
}

CRtThreadManager* CRtThreadManager::Instance()
{
//	RT_ASSERTE(s_pThreadManagerOnlyOne);
	if (!s_pThreadManagerOnlyOne) {
//		RT_ERROR_TRACE("CRtThreadManager::Instance, s_pThreadManagerOnlyOne is NULL!"
//			" You should alloc it in the stack or heap first!");

		// We have to new <CRtThreadManager> becauase Connect1.1 UI doesn't
		// alloc it in the main function.
		// we have to assume it's in main thread.
		RT_INFO_TRACE("CRtThreadManager::Instance, new CRtThreadManager.");
		new CRtThreadManager();
		s_bThreadManagerOnlyOneByNew = TRUE;

		RtResult rv = s_pThreadManagerOnlyOne->InitMainThread(0, NULL);
		if (RT_FAILED(rv)) {
			delete s_pThreadManagerOnlyOne;
			s_pThreadManagerOnlyOne = NULL;
			return NULL;
		}
		else {
#if (defined RT_LINUX || defined RT_UNIX ) && defined RT_PORT_CLIENT
			//Now LINUX and SOLARIS client all used load model, so can need not register the function,
			//otherwise if unload lib before exit, will cause crash 2005 11/17
#else
			int nRet = ::atexit(CleanupOnlyOne);
			if (nRet != 0) {
				RT_ERROR_TRACE("CRtThreadManager::Instance, atexit() failed. err=" << errno);
			}
#endif
		}
	}
	return s_pThreadManagerOnlyOne;
}

RtResult CRtThreadManager::InitMainThread(int aArgc, char** aArgv)
{
	RT_INFO_TRACE_THIS("CRtThreadManager::InitMainThread,");

	{
		CRtMutexGuardT<MutexType> theGuard(m_Mutex);
		RT_ASSERTE(m_Threads.empty());
		if (!m_Threads.empty()) {
			RT_ERROR_TRACE("CRtThreadManager::InitMainThread, You should InitMainThread before creating other thread!");
			return RT_ERROR_ALREADY_INITIALIZED;
		}
	}

#ifdef RT_HAS_BUILTIN_ATOMIC_OP
	CRtAtomicOperationT<CRtMutexThread>::init_functions();
#endif // RT_HAS_BUILTIN_ATOMIC_OP

	RtResult rv = SocketStartup();
	if (RT_FAILED(rv))
		return rv;
	
	IRtReactor *pReactorMain = NULL;
#ifdef RT_WIN32
	if (GetNetworkThreadModule() == TM_SINGLE_MAIN)
		pReactorMain = CreateNetworkReactor();
	else
		pReactorMain = new CRtReactorWin32Message();
#else
	if (GetNetworkThreadModule() == TM_SINGLE_MAIN)
		pReactorMain = CreateNetworkReactor();
	else {
		// the main thread on Linux is task thread.
		pReactorMain = NULL;
	}
#endif // RT_WIN32

	CRtThread *pThread = NULL;
	if (pReactorMain)
		rv = CreateReactorThread(CRtThreadManager::TT_MAIN, pReactorMain, pThread);
	else {
#ifdef RT_MACOS
		pThread = new CRtThreadMacEventPatch();
#elif defined (RT_PORT_CLIENT) || defined (RT_HEARTBEAT_SUPPORT)
		pThread = new CRtThreadHeartBeat();
#else
		pThread = new CRtThreadTask();
#endif // RT_MACOS
		if (pThread)
			rv = pThread->Create(CRtThreadManager::TT_MAIN);
		else
			rv = RT_ERROR_OUT_OF_MEMORY;
	}

	if (RT_FAILED(rv)) {
		if (pThread)
		pThread->Destory(rv);
		return rv;
	}

	// create network thread when program startuping.
	rv = SpawnNetworkThread_i();
	if (RT_FAILED(rv)) {
		if (pThread)
		pThread->Destory(RT_OK);
		return RT_ERROR_UNEXPECTED;
	}

#if 0
#ifdef RT_WIN32
	RT_INFO_TRACE_THIS("CRtThreadManager::InitMainThread, SetThreadPriority for main thread and network thread.");
	BOOL bSuccess = ::SetThreadPriority(::GetCurrentThread(), THREAD_PRIORITY_TIME_CRITICAL);
	if (!bSuccess)
		RT_WARNING_TRACE_THIS("CRtThreadManager::InitMainThread, SetThreadPriority for main thread failed!");
	bSuccess = ::SetThreadPriority(m_pThreadNetwork->GetThreadHandle(), THREAD_PRIORITY_TIME_CRITICAL);
	if (!bSuccess)
		RT_WARNING_TRACE_THIS("CRtThreadManager::InitMainThread, SetThreadPriority for network thread failed!");
#endif
#endif

	pThread->OnThreadInit();
	return RT_OK;
}

IRtReactor* CRtThreadManager::CreateNetworkReactor()
{
	IRtReactor *pReactorRet = NULL;
#ifdef RT_USE_REACTOR_SELECT
	pReactorRet = new CRtReactorSelect();
#elif defined (RT_WIN32)
	pReactorRet = new CRtReactorWin32AsyncSelect();
#else // !RT_WIN32
	struct utsname utName;
	if (::uname(&utName) < 0) {
		RT_ERROR_TRACE("CRtThreadManager::CreateNetworkReactor, uname() failed!"
			"err=" << errno);
	}
	else {
		if (!strncmp(utName.release, "2.4.", 4)) {
			int nMinVer = ::atoi(utName.release + 4);
			if (nMinVer >= 18)
				pReactorRet = new CRtReactorRealTimeSignal();
			else {
				RT_ERROR_TRACE("CRtThreadManager::CreateNetworkReactor,"
					" Only 2.4.18 or above supports realtime signal."
					" release=" << utName.release << 
					" sysname=" << utName.sysname);
			}
		}
#ifdef RT_ENABLE_EPOLL
		else if (!strncmp(utName.release, "2.6.", 4)) {
			pReactorRet = new CRtReactorEpoll();
		}
#endif // RT_ENABLE_EPOLL
		else {
			RT_ERROR_TRACE("CRtThreadManager::CreateNetworkReactor,"
				" don't support this release of Linux."
				" release=" << utName.release << 
				" sysname=" << utName.sysname);
		}
	}
#endif // RT_USE_REACTOR_SELECT
	
	return pReactorRet;
}

RtResult CRtThreadManager::SpawnNetworkThread_i()
{
	if (m_pThreadNetwork)
		return RT_OK;

	TModule tmodule = GetNetworkThreadModule();
	if (tmodule == TM_SINGLE_MAIN) {
		std::auto_ptr<CRtThreadDummy> pThreadReactor(new CRtThreadDummy());
		if (!pThreadReactor.get())
			return RT_ERROR_OUT_OF_MEMORY;

		RtResult rv = pThreadReactor->Init(GetThread(TT_MAIN), TT_NETWORK);
		if (RT_FAILED(rv))
			return rv;

		m_pThreadNetwork = pThreadReactor.release();
	}
	else {
		IRtReactor *pReactorNetwork = CreateNetworkReactor();
		RtResult rv = CreateReactorThread(
			TT_NETWORK, pReactorNetwork, m_pThreadNetwork);
		if (RT_FAILED(rv))
			return rv;
	}

	return RT_OK;
}

#if 0
RtResult CRtThreadManager::CreateUserReactorThread(CRtThread *&aThread)
{
	RT_ASSERTE(!aThread);

	IRtReactor *pReactor = NULL;
#ifdef RT_WIN32
	pReactor = new CRtReactorWin32Message();
#else
	pReactor = NULL;
#endif // RT_LINUX
	RT_ASSERTE_RETURN(pReactor, RT_ERROR_OUT_OF_MEMORY);

	RtResult rv = CreateReactorThread(GetAndIncreamUserType(), pReactor, aThread);
	return rv;
}
#endif

RtResult CRtThreadManager::CreateUserTaskThread(CRtThread *&aThread, TFlag aFlag, BOOL bWithTimerQueue)
{
	RT_ASSERTE(!aThread);

	CRtThread *pThread = NULL;
	if (bWithTimerQueue)
		pThread = new CRtThreadTask();
	else
		pThread = new CRtThreadTaskWithEventQueueOnly();
	RT_ASSERTE_RETURN(pThread, RT_ERROR_OUT_OF_MEMORY);

	RtResult rv = pThread->Create(GetAndIncreamUserType(), aFlag);
	if (RT_FAILED(rv)) {
		pThread->Destory(rv);
		return rv;
	}

	aThread = pThread;
	return RT_OK;
}

CRtThreadManager::TType CRtThreadManager::GetAndIncreamUserType()
{
	CRtMutexGuardT<MutexType> theGuard(m_Mutex);
	return ++m_TTpyeUserThread;
}

RtResult CRtThreadManager::CreateReactorThread(TType aType, IRtReactor *aReactor, CRtThread *&aThread)
{
	// Note: aReactor will be delete if failed.
	RtResult rv;
	RT_ASSERTE_RETURN(aReactor, RT_ERROR_INVALID_ARG);
	RT_ASSERTE(!aThread);

	std::auto_ptr<CRtThreadReactor> pThreadReactor(new CRtThreadReactor());
	if (!pThreadReactor.get()) {
		delete aReactor;
		return RT_ERROR_OUT_OF_MEMORY;
	}
	
	rv = pThreadReactor->Init(aReactor);
	if (RT_FAILED(rv))
		return rv;

	rv = pThreadReactor->Create(aType);
	if (RT_FAILED(rv))
		return rv;

	aThread = pThreadReactor.release();
	return RT_OK;
}

RtResult CRtThreadManager::RegisterThread(CRtThread* aThread)
{
	RT_ASSERTE_RETURN(aThread, RT_ERROR_INVALID_ARG);
	CRtMutexGuardT<MutexType> theGuard(m_Mutex);

	ThreadsType::iterator iter = m_Threads.begin();
	for ( ; iter != m_Threads.end(); ++iter) {
		// we don't check equal ThreadId because mutil CRtThreads may use the same id.
		if ((*iter) == aThread || 
			(*iter)->GetThreadType() == aThread->GetThreadType()) 
		{
			RT_ERROR_TRACE("CRtThreadManager::RegisterThread, have registered."
				" aThread=" << aThread <<
				" tid=" << aThread->GetThreadId() << 
				" type=" << aThread->GetThreadType());
			RT_ASSERTE(FALSE);
			return RT_ERROR_FOUND;
		}
	}

	m_Threads.push_back(aThread);
	return RT_OK;
}

RtResult CRtThreadManager::UnregisterThread(CRtThread* aThread)
{
	RT_ASSERTE_RETURN(aThread, RT_ERROR_INVALID_ARG);
	CRtMutexGuardT<MutexType> theGuard(m_Mutex);

	RT_INFO_TRACE_THIS("CRtThreadManager::UnregisterThread, thread="<<aThread);
	ThreadsType::iterator iter = m_Threads.begin();
	for ( ; iter != m_Threads.end(); ++iter) {
		if ((*iter) == aThread || 
			(*iter)->GetThreadType() == aThread->GetThreadType()) 
		{
			m_Threads.erase(iter);
			return RT_OK;
		}
	}
	return RT_ERROR_NOT_FOUND;
}
void CRtThreadManager::GetSingletonMutex(CRtMutexThreadRecursive *&aMutex)
{
	aMutex = &m_SingletonMutex;
}
RtResult CRtThreadManager::StopAllThreads(CRtTimeValue* aTimeout)
{
	// not support timeout now.
	RT_ASSERTE(!aTimeout);
	
	// only the main thread can stop all threads.
	CRtThread *pMain = GetThread(TT_MAIN);
	if (pMain) {
		RT_ASSERTE_RETURN(
			IsThreadEqual(pMain->GetThreadId(), GetThreadSelfId()), 
			RT_ERROR_FAILURE);
	}

	ThreadsType tmpThreads;
	{
		CRtMutexGuardT<MutexType> theGuard(m_Mutex);
		tmpThreads = m_Threads;
	}
	ThreadsType::iterator iter = tmpThreads.begin();
	for ( ; iter != tmpThreads.end(); ++iter) {
		if ((*iter)->GetThreadType() != TT_MAIN)
			(*iter)->Stop(aTimeout);
	}

	return RT_OK;
}

RtResult CRtThreadManager::JoinAllThreads()
{
	// only the main thread can Join all threads.
	CRtThread *pMain = GetThread(TT_MAIN);
	if (pMain) {
		RT_ASSERTE_RETURN(
			IsThreadEqual(pMain->GetThreadId(), GetThreadSelfId()), 
			RT_ERROR_FAILURE);
	}

	ThreadsType tmpThreads;
	{
		CRtMutexGuardT<MutexType> theGuard(m_Mutex);
		tmpThreads = m_Threads;
	}
	ThreadsType::iterator iter = tmpThreads.begin();
	for ( ; iter != tmpThreads.end(); ++iter) {
		if ((*iter)->GetThreadType() != TT_MAIN)
			(*iter)->Join(RT_THREAD_JOIN_TIMEOUT);
	}

	return RT_OK;
}

CRtThread* CRtThreadManager::GetThread(TType aType)
{
	CRtMutexGuardT<MutexType> theGuard(m_Mutex);
	if (TT_CURRENT == aType) {
		RT_THREAD_ID tidSelf = GetThreadSelfId();
		ThreadsType::iterator iter = m_Threads.begin();
		for ( ; iter != m_Threads.end(); ++iter) {
			if ((*iter)->GetThreadId() == tidSelf)
				return (*iter);
		}
	}
	else {
		ThreadsType::iterator iter = m_Threads.begin();
		for ( ; iter != m_Threads.end(); ++iter) {
			if ((*iter)->GetThreadType() == aType)
				return (*iter);
		}
	}
	return NULL;
}

IRtReactor* CRtThreadManager::GetThreadReactor(TType aType)
{
	CRtThread *pThread = GetThread(aType);
	if (pThread)
		return pThread->GetReactor();
	else
		return NULL;
}

IRtEventQueue* CRtThreadManager::GetThreadEventQueue(TType aType)
{
	CRtThread *pThread = GetThread(aType);
	if (pThread)
		return pThread->GetEventQueue();
	else
		return NULL;
}

IRtTimerQueue* CRtThreadManager::GetThreadTimerQueue(TType aType)
{
	CRtThread *pThread = GetThread(aType);
	if (pThread)
		return pThread->GetTimerQueue();
	else
		return NULL;
}

BOOL CRtThreadManager::IsEqualCurrentThread(TType aType)
{
	CRtThread *pThread = Instance()->GetThread(aType);
	if (pThread)
		return IsThreadEqual(pThread->GetThreadId(), GetThreadSelfId());
	else
		return FALSE;
}

void CRtThreadManager::SleepMs(DWORD aMsec)
{
#ifdef RT_WIN32
	::Sleep(aMsec);
#else
	struct timespec ts, rmts;
	ts.tv_sec = aMsec/1000;
	ts.tv_nsec = (aMsec%1000)*1000000;

	for ( ; ; ) {
		int nRet = ::nanosleep(&ts, &rmts);
		if (nRet == 0)
			break;
		if (errno == EINTR) {
			ts = rmts;
		}
		else {
			RT_ERROR_TRACE("CRtThreadManager::SleepMs,"
				"nanosleep() failed! err=" << errno);
			break;
		}
	}
#endif // RT_WIN32
}

#ifdef RT_WIN32
BOOL CRtThreadManager::s_bSocketInited = FALSE;
#endif // RT_WIN32

RtResult CRtThreadManager::SocketStartup()
{
#ifdef RT_WIN32
	if (s_bSocketInited)
		return RT_OK;
	s_bSocketInited = TRUE;

	WSADATA wsaData;
	WORD wVersionRequested = MAKEWORD(2, 2);
	if (::WSAStartup(wVersionRequested, &wsaData) != 0) {
		RT_ERROR_TRACE("CRtThreadManager::SocketStartup, WSAStartup() failed!"
			" err=" << ::WSAGetLastError());
		s_bSocketInited = FALSE;
		return RT_ERROR_FAILURE;
	}

	if (LOBYTE(wsaData.wVersion) != 2 || HIBYTE(wsaData.wVersion) != 2) {
		RT_ERROR_TRACE("CRtThreadManager::SocketStartup, version error!"
			" wsaData.wVersion=" << wsaData.wVersion <<
			" wsaData.wVersion=" << wsaData.wVersion <<
			" err=" << ::WSAGetLastError());
		SocketCleanup();
		return RT_ERROR_NOT_AVAILABLE;
	}
#endif // RT_WIN32
	return RT_OK;
}

RtResult CRtThreadManager::SocketCleanup()
{
#ifdef RT_WIN32
	if (!s_bSocketInited)
		return RT_OK;
	s_bSocketInited = FALSE;

	if (::WSACleanup() != 0) {
		RT_ERROR_TRACE("CRtThreadManager::SocketCleanup, WSACleanup() failed!"
			" err=" << ::WSAGetLastError());
		return RT_ERROR_UNEXPECTED;
	}
#endif // RT_WIN32
	return RT_OK;
}

//#define RT_NETWORK_THREAD_SINGLE_MAIN

CRtThreadManager::TModule CRtThreadManager::GetNetworkThreadModule()
{
#ifdef RT_NETWORK_THREAD_SINGLE_MAIN
	return TM_SINGLE_MAIN;
#else
	return TM_MULTI_ONE_DEDICATED;
#endif // RT_NETWORK_THREAD_SINGLE_MAIN
}

#ifdef RT_WIN32
BOOL WINAPI DllMain(HANDLE hinstDLL, DWORD fdwReason, LPVOID lpvReserved)
{
	switch(fdwReason)
    {
	case DLL_PROCESS_ATTACH :
		RT_INFO_TRACE("rtframework version: BOX 4200, 413, 0, 0");
		break;

	case DLL_THREAD_ATTACH :
		break;

	case DLL_THREAD_DETACH :
		break;

	case DLL_PROCESS_DETACH :
		RT_INFO_TRACE("rtframework DLL_PROCESS_DETACH");
		break;

	default :
		RT_ASSERTE(FALSE);
		break;
	}
    return TRUE;
}
#endif // RT_WIN32

